package com.cloudcentral.websocket;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.TypeReference;
import com.cloudcentral.entity.DetectResult;
import com.cloudcentral.entity.Radio;
import com.cloudcentral.entity.RadioDevice;
import com.cloudcentral.enums.RadioStatus;
import com.cloudcentral.exception.FailureException;
import com.cloudcentral.service.*;
import com.cloudcentral.util.RSAUtil;
import com.cloudcentral.util.SpringContextUtils;
import com.cloudcentral.websocket.entity.WSData;
import com.cloudcentral.websocket.entity.WSMessage;
import com.cloudcentral.websocket.entity.dto.RadioStatusDTO;
import jakarta.websocket.*;
import jakarta.websocket.server.ServerEndpoint;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RLock;
import org.redisson.api.RSet;
import org.redisson.api.RedissonClient;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import java.net.URLDecoder;
import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArraySet;
import java.util.concurrent.TimeUnit;

import static com.cloudcentral.util.Utils.errorMessage;

/**
 * Create by Administrator
 * 2023/12/1 17:58
 */
@Slf4j
@Component
@EnableScheduling
@ServerEndpoint("/ws/radio")
public class RadioWebSocket {
  private static final List<Object> beans = new ArrayList<>();
  // concurrent包的线程安全Set,用来存放每个客户端对应的WebSocket对象。
  private static final CopyOnWriteArraySet<RadioWebSocket> webSocketSet = new CopyOnWriteArraySet<>();
  // session集合,存放对应的session
  private static final ConcurrentHashMap<Integer, Session> sessionPool = new ConcurrentHashMap<>();
  private Integer radioId;
  // 与某个客户端的连接会话，需要通过它来给客户端发送数据
  private Session session;
  // 信号间隔最大时长
  public static final Duration duration = Duration.ofSeconds(15);
  // 上次通信时间戳
  public long lastTime;

  private static <T> T bean(Class<T> clazz) {
    T bean = (T) beans.stream().filter(item -> item.getClass().equals(clazz)).findAny().orElse(null);
    if (bean == null) {
      bean = SpringContextUtils.getBean(clazz);
      beans.add(bean);
    }
    return bean;
  }

  private void done() throws Exception {
    session.close();
    sessionPool.remove(radioId);
    webSocketSet.remove(this);
    bean(RadioOnlineService.class).status(radioId, false);
  }

  @Scheduled(initialDelay = 0, fixedRate = 1000)
  private void clearDone() {
    try {
      List<RadioWebSocket> list = webSocketSet.stream().filter(item -> System.currentTimeMillis() - item.lastTime > duration.toMillis()).toList();
      if (!list.isEmpty()) for (RadioWebSocket radioWebSocket : list) radioWebSocket.done();
    } catch (Exception e) {
      log.error("RadioWebSocket清理异常", e);
    }
  }

  @OnOpen
  public void onOpen(Session session) throws Exception {
    this.session = session;
    lastTime = System.currentTimeMillis();
    String message = session.getQueryString();
    message = URLDecoder.decode(message, StandardCharsets.UTF_8);
    message = RSAUtil.privateKey().decrypt(message);
    Radio radio = JSON.parseObject(message, Radio.class);
    log.info("RadioWebSocket建立连接中,ID：{}", radio.getId());
    RLock lock = bean(RedissonClient.class).getLock("WebSocket-Open");
    try {
      lock.tryLock(10, 10, TimeUnit.SECONDS);
      radio = bean(RadioService.class).register(radio);
      if (radio == null) throw new FailureException("注册失败");
      else {
        Session historySession = sessionPool.get(radio.getId());
        // historySession不为空,说明已经有台站注册,应该删除登陆的WebSocket对象
        if (historySession != null) {
          webSocketSet.removeIf((item) -> item.session.equals(historySession));
          historySession.close();
        }
        // 建立连接
        this.radioId = radio.getId();
        webSocketSet.add(this);
        sessionPool.put(radio.getId(), session);
        sendMessage(session, WSMessage.ok(WSMessage.WSMessageType.REGISTER, radio));
        bean(RadioOnlineService.class).status(radio.getId(), true);
        log.info("RadioWebSocket建立连接成功,当前连接数为：{}", webSocketSet.size());
      }
    } finally {
      lock.unlock();
    }
  }

  @OnError
  public void onError(Session session, Throwable e) {
    log.error("RadioWebSocket发生错误：{}", errorMessage(e));
    sendMessage(session, WSMessage.error(errorMessage(e)));
  }

  @OnClose
  public void onClose(Session session) throws Exception {
    done();
    log.info("RadioWebSocket连接断开,当前连接数为：{}", webSocketSet.size());
  }

  @OnMessage(maxMessageSize = 1024 * 1024 * 8)
  public void onMessage(Session session, String message) throws Exception {
    lastTime = System.currentTimeMillis();
    RLock lock = bean(RedissonClient.class).getLock("WebSocket-OnMessage");
    try {
      lock.tryLock(10, 10, TimeUnit.SECONDS);
      message = RSAUtil.privateKey().decrypt(message);
      log.info("RadioWebSocket收到文本消息：{}", message);
      WSData<String> wsData = JSON.parseObject(message, new TypeReference<>() {
      });
      //心跳消息
      if (wsData.getType().equalsIgnoreCase(WSMessage.WSMessageType.HEART.getType())) sendMessage(session, WSMessage.heart());
      //客户端回传探测任务是否启动成功
      else if (wsData.getType().equalsIgnoreCase(WSMessage.WSMessageType.START_DETECT.getType())) {
        JSONObject data = JSON.parseObject(wsData.getData());
        Long detectId = data.getLong("detectId");
        Boolean status = data.getBoolean("status");
        if (status) {
          RSet<Long> rSet = bean(RedissonClient.class).getSet("WS-Detect");
          rSet.tryAdd(detectId);
          rSet = bean(RedissonClient.class).getSet("WS-Detect");
          CloudWebSocket.sendAllMessage(WSMessage.ok(WSMessage.WSMessageType.DETECT_RUNNING, rSet.stream().toList()));
        } else CloudWebSocket.sendMessageForCurrentUser(WSMessage.fail(WSMessage.WSMessageType.START_DETECT, data));
      }
      //客户端回传探测任务是否停止成功
      else if (wsData.getType().equalsIgnoreCase(WSMessage.WSMessageType.STOP_DETECT.getType())) {
        JSONObject data = JSON.parseObject(wsData.getData());
        Long detectId = data.getLong("detectId");
        Boolean status = data.getBoolean("status");
        if (status) {
          RSet<Long> rSet = bean(RedissonClient.class).getSet("WS-Detect");
          rSet.remove(detectId);
          rSet = bean(RedissonClient.class).getSet("WS-Detect");
          CloudWebSocket.sendAllMessage(WSMessage.ok(WSMessage.WSMessageType.DETECT_RUNNING, rSet.stream().toList()));
        } else CloudWebSocket.sendMessageForCurrentUser(WSMessage.fail(WSMessage.WSMessageType.STOP_DETECT, data));
      }
      //台站状态
      else if (wsData.getType().equalsIgnoreCase(WSMessage.WSMessageType.RADIO_STATUS.getType())) {
        RadioStatusDTO dto = JSON.parseObject(wsData.getData(), RadioStatusDTO.class);
        if (dto.getStatus() != null) RadioStatus.value(radioId, RadioStatus.valueOf(dto.getStatus()));
        RadioDevice radioDevice = bean(RadioDeviceService.class).submit(RadioDevice.builder()
            .radioId(radioId)
            .temperature(dto.getTemperature())
            .electricCurrent(dto.getElectricCurrent())
            .frontVoltage(dto.getFrontVoltage())
            .backVoltage(dto.getBackVoltage())
            .frequency(dto.getFrequency())
            .power(dto.getPower())
            .build());
        JSONObject json = JSON.parseObject(JSON.toJSONString(radioDevice));
        json.put("status", dto.getStatus());
        CloudWebSocket.sendAllMessage(WSMessage.ok(WSMessage.WSMessageType.RADIO_STATUS, json));
      }
      //上报PCM数据流
      else if (wsData.getType().equalsIgnoreCase(WSMessage.WSMessageType.PCM.getType())) bean(PCMService.class).submit(JSON.parseObject(wsData.getData()));
      //上传探测数据-探测数据
      else if (wsData.getType().equalsIgnoreCase(WSMessage.WSMessageType.DETECT_DATA.getType())) {
        try {
          DetectResult detectResult = JSON.parseObject(wsData.getData(), DetectResult.class);
          bean(DetectResultService.class).submit(detectResult);
        } catch (Exception e) {
          log.error("提交探测数据失败", e);
        }
      }
    } finally {
      lock.unlock();
    }
  }

  public static <T> void sendAllMessage(WSMessage<T> wsMessage) {
    for (RadioWebSocket webSocket : webSocketSet) sendMessage(webSocket.session, wsMessage);
  }

  public static <T> void sendMessageById(Integer id, WSMessage<T> wsMessage) {
    sendMessage(sessionPool.get(id), wsMessage);
  }

  public static <T> void sendMessage(Session session, WSMessage<T> wsMessage) {
    if (session != null) {
      String message = JSON.toJSONString(wsMessage);
      RLock lock = bean(RedissonClient.class).getLock("WebSocket-SendMessage");
      try {
        lock.tryLock(10, 10, TimeUnit.SECONDS);
        message = RSAUtil.privateKey().encrypt(message);
        session.getBasicRemote().sendText(message);
        log.info("RadioWebSocket发送消息: {}", message);
      } catch (Exception e) {
        log.error("RadioWebSocket发送消息失败", e);
      } finally {
        lock.unlock();
      }
    }
  }
}
